package com.atguigu.transfrom;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author gmd
 * @desc Flatmap映射算子介绍
 * @since 2024-11-25 09:54:29
 */
public class FlatmapDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3)
        );

        /*
         * FlatMap示例：一进多出（包含0出）
         * 该示例演示了FlatMap函数的使用，实现一进多出（包含0出）的效果。
         * 对于s1的数据，一进一出；对于s2的数据，一进2出；对于s3的数据，一进0出（类似于过滤的效果）。
         * 与Map函数不同，Map函数通过return控制一进一出；而FlatMap函数通过Collector来输出，调用几次就输出几条。
         */
        SingleOutputStreamOperator<String> flatmap = sensorDS.flatMap(new FlatMapFunction<WaterSensor, String>() {
            @Override
            public void flatMap(WaterSensor value, Collector<String> out) throws Exception {
                if ("s1".equals(value.getId())) {
                    // 如果是 s1，输出 vc
                    out.collect(value.getId()+"-"+value.getVc());
                } else if ("s2".equals(value.getId())) {
                    // 如果是 s2，分别输出ts和vc
                    out.collect(value.getId()+"-"+value.getTs());
                    out.collect(value.getId()+"--"+value.getVc());
                }
            }
        });

        flatmap.print();
        env.execute();
    }

}
